package com.qingcloud.df.executor.cluster;
import cn.hutool.extra.spring.SpringUtil;
import com.qingcloud.base.remote.executor.ExecutorParam;
import com.qingcloud.df.sdk.component.enums.NodeTypeEnum;
import com.qingcloud.df.sdk.component.message.ExecutorMetrics;
import com.qingcloud.df.sdk.component.message.NodeInfo;
import lombok.extern.log4j.Log4j2;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;

@Log4j2
@Component
public class HealthService {


    public void checkServer() {

    }

    public void checkNode() {
        Map<String,NodeInfo> nodeInfoMap = ClusterManager.getNodeMap();
        if (Objects.isNull(nodeInfoMap)) {
            log.warn("Health check ignore, no executor need check.");
            return;
        }
        nodeInfoMap.forEach((nodeId,nodeInfo)->{
            if(nodeInfo.getNodeType().equalsIgnoreCase(NodeTypeEnum.EXECUTOR.getName())){
                doCheck(nodeInfo);
            }
        });
    }

    private void doCheck(NodeInfo nodeInfo){
        boolean timeout =  nodeInfo.timeout();
        //执行器节点不可达判断方式暂时采用60s 超时判断，无法确保完全正确，需要迭代增加可靠性机制  TODO
        if(timeout) {
            log.warn("executor time [{}] timeout ,cluster will remove it and reassign jobs", nodeInfo.getAkkaAddress());
            ClusterManager.removeNode(nodeInfo.getAkkaAddress());
            ClusterManager.removeNodeJobs(nodeInfo.getAkkaAddress());
            if(nodeInfo.getExecutorMetrics()!=null){
                ExecutorMetrics metrics= nodeInfo.getExecutorMetrics();
                List<Integer> jobList  = metrics.getJobList();
                String env  =nodeInfo.getEnv();
                List<ExecutorParam.JobParam> jobParams = new ArrayList<>();
                for(Integer jobId : jobList){
                    ExecutorParam.JobParam jobParam = new ExecutorParam.JobParam();
                    jobParam.setJobId(jobId);
                    jobParam.setEnv(env);
                    jobParams.add(jobParam);
                }
                SpringUtil.getBean(JobService.class).assignJobs(jobParams);
            }
        }
    }
}